前情提要

上一篇分享了三個用 API 呼叫的 LLMs，那除了單純的跟 AI 溝通，如果要串接其他 AI 工具或功能，像是 Memory、Retrieve 甚至是更進階的 Agents，自己一點一點慢慢刻當然是做得到。但是有個框架可以幫助你將
這些工具、元件、語言模型都整合起來，何樂而不為呢？那這就是我們今天要分享的主角 - LangChain

LangChain 🦜️🔗

簡單來說，LangChain 整合了大多數的語言模型，像是 OpenAI、Gemini、Ollama 等。但如果沒有的話，像是台智雲，也可以自己將模型整合至 LangChain 架構裡面，後面會實戰程式！其實跟同事實作的時候，LangChain 的評價很兩極，有些人覺得都整合起來很方便，有些人覺得還要多學一個框架很麻煩。我自己是覺得都挺有道理，畢竟剛開始用 LangChain 確實花了一段時間跟他混熟。而且從我知道 LangChain 到現在，不到一年的時間從 0.1.6 版到今天 (2024/08/16) 0.2.14 版，之前還曾經一個禮拜更新兩次，所以搞不好一年後或幾個月後這篇文章的程式全部都不能用了 🤣

LangChain Package

Google Gemini - 實戰🔥

要安裝 langchain-google-genai 這個套件，套件說明連結：langchain-google-genai

# 匯入需要的套件
from dotenv import load_dotenv
from langchain_google_genai import ChatGoogleGenerativeAI
load_dotenv()
# 選擇模型和想要調整的參數
llm = ChatGoogleGenerativeAI(model="gemini-pro", temperature=0.2)
response = llm.invoke("Aespa的成員有誰？")
print(response.content)

程式碼結果探討 🧐：

• 用了 Langchain 之後，這樣子或許沒什麼感覺有比較精簡，但隨著慢慢了解他，就會慢慢懂他的好了～簡單指定模型和設定參數，然後輸入內容就可以得到 AI 的回覆了

OpenAI - 實戰🔥

要安裝 langchain-openai 這個套件，套件說明連結：langchain-openai

# 匯入需要的套件
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
# 選擇模型和想要調整的參數
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
response = llm.invoke("Aespa的成員有誰？")
print(response.content)

程式碼結果探討 🧐：

• 與 Gemini 的部分相比，不知道大家有沒有發現除了匯入模型的函數不一樣，其他輸入參數的部分都是大同小異。我自己覺得這是最快理解 LangChain 框架的方式，那這只是個開始，後續會再慢慢介紹 LangChain 的強大之處。

台智雲 🇹🇼 - 實戰🔥

要安裝 langchain-core & Requests 這兩個套件，套件說明連結：LangChain Custom LLM

• 先封裝台智雲的 LLM，接著就可以在 LangChain 架構上使用特定 LLM，可以參考台智雲官網：台智雲 Conversation API

import os
import json
import requests
from pydantic.v1 import Field
from typing import Any, Dict, List, Mapping, Optional, Tuple
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.language_models.base import BaseLanguageModel
from langchain_core.messages.base import BaseMessage
from langchain_core.outputs.chat_generation import ChatGeneration
from langchain_core.outputs.chat_result import ChatResult
from langchain_core.messages.chat import ChatMessage
from langchain_core.messages.ai import AIMessage
from langchain_core.messages.human import HumanMessage
from langchain_core.messages.system import SystemMessage
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from dotenv import load_dotenv
load_dotenv()

class _ChatFormosaFoundationCommon(BaseLanguageModel):
    base_url: str = "https://api-ams.twcc.ai/api"
    """Base url the model is hosted under."""

    model: str = "ffm-llama3-70b-chat"
    """Model name to use."""
    
    temperature: Optional[float] = 0.01
    """The temperature of the model. Increasing the temperature will
    make the model answer more creatively."""

    stop: Optional[List[str]]
    """Sets the stop tokens to use."""

    top_k: int = 10
    """Reduces the probability of generating nonsense. A higher value (e.g. 100)
    will give more diverse answers, while a lower value (e.g. 10)
    will be more conservative. (Default: 50)"""

    top_p: float = 1
    """Works together with top-k. A higher value (e.g., 0.95) will lead
    to more diverse text, while a lower value (e.g., 0.5) will
    generate more focused and conservative text. (Default: 1)"""

    max_new_tokens: int = 1024
    """The maximum number of tokens to generate in the completion.
    -1 returns as many tokens as possible given the prompt and
    the models maximal context size."""

    frequence_penalty: float = 1.03
    """Penalizes repeated tokens according to frequency."""

    model_kwargs: Dict[str, Any] = Field(default_factory=dict)
    """Holds any model parameters valid for `create` call not explicitly specified."""
    try:
        ffm_api_key: Optional[str] = os.environ['FFM_API_KEY']
    except:
        ffm_api_key: Optional[str] = None
        
    @property
    def _default_params(self) -> Dict[str, Any]:
        """Get the default parameters for calling FFM API."""
        normal_params = {
            "temperature": self.temperature,
            "max_new_tokens": self.max_new_tokens,
            "top_p": self.top_p,
            "frequence_penalty": self.frequence_penalty,
            "top_k": self.top_k,
        }
        return {**normal_params, **self.model_kwargs}

    def _call(
        self,
        prompt,
        stop: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> str:
        if self.stop is not None and stop is not None:
            raise ValueError("`stop` found in both the input and default params.")
        elif self.stop is not None:
            stop = self.stop
        elif stop is None:
            stop = []
        params = {**self._default_params, "stop": stop, **kwargs}
        parameter_payload = {"parameters": params, "messages": prompt, "model": self.model}

        # HTTP headers for authorization
        headers = {
            "X-API-KEY": self.ffm_api_key,
            "X-API-HOST": "afs-inference",
            "Content-Type": "application/json",
        }
        endpoint_url = f"{self.base_url}/models/conversation"
        # send request
        try:
            response = requests.post(
                url=endpoint_url, 
                headers=headers, 
                data=json.dumps(parameter_payload, ensure_ascii=False).encode("utf8"),
                stream=False,
            )
            response.encoding = "utf-8"
            generated_text = response.json()
            if response.status_code != 200:
                detail = generated_text.get("detail")
                raise ValueError(
                    f"FormosaFoundationModel endpoint_url: {endpoint_url}\n"
                    f"error raised with status code {response.status_code}\n"
                    f"Details: {detail}\n"
                )

        except requests.exceptions.RequestException as e:  # This is the correct syntax
            raise ValueError(f"FormosaFoundationModel error raised by inference endpoint: {e}\n")

        if generated_text.get("detail") is not None:
            detail = generated_text["detail"]
            raise ValueError(
                f"FormosaFoundationModel endpoint_url: {endpoint_url}\n"
                f"error raised by inference API: {detail}\n"
            )
        
        if generated_text.get("generated_text") is None:
            raise ValueError(
                f"FormosaFoundationModel endpoint_url: {endpoint_url}\n"
                f"Response format error: {generated_text}\n"
            )

        return generated_text

class ChatFormosaFoundationModel(BaseChatModel, _ChatFormosaFoundationCommon):
    """`FormosaFoundation` Chat large language models API.

    The environment variable ``OPENAI_API_KEY`` set with your API key.

    Example:
        .. code-block:: python
            ffm = ChatFormosaFoundationModel(model_name="llama2-7b-chat-meta")
    """

    @property
    def _llm_type(self) -> str:
        return "ChatFormosaFoundationModel"

    @property
    def lc_serializable(self) -> bool:
        return True

    def _convert_message_to_dict(self, message: BaseMessage) -> dict:
        if isinstance(message, ChatMessage):
            message_dict = {"role": message.role, "content": message.content}
        elif isinstance(message, HumanMessage):
            message_dict = {"role": "human", "content": message.content}
        elif isinstance(message, AIMessage):
            message_dict = {"role": "assistant", "content": message.content}
        elif isinstance(message, SystemMessage):
            message_dict = {"role": "system", "content": message.content}
        else:
            raise ValueError(f"Got unknown type {message}")
        return message_dict

    def _create_conversation_messages(
        self,
        messages: List[BaseMessage], 
        stop: Optional[List[str]]
    ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
        params: Dict[str, Any] = {**self._default_params}

        if stop is not None:
            if "stop" in params:
                raise ValueError("`stop` found in both the input and default params.")
            params["stop"] = stop

        message_dicts = [self._convert_message_to_dict(m) for m in messages]
        return message_dicts, params

    def _create_chat_result(self, response: Mapping[str, Any]) -> ChatResult:
        chat_generation = ChatGeneration(
            message = AIMessage(content=response.get("generated_text")),
            generation_info = {
                "token_usage": response.get("generated_tokens"), 
                "model": self.model
            }
        )

        return ChatResult(generations=[chat_generation])

    def _generate(
        self,         
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {**params, **kwargs}
        response = self._call(prompt=message_dicts)
        if type(response) is str: # response is not the format of dictionary
            return response

        return self._create_chat_result(response)

    async def _agenerate(
        self, messages: List[BaseMessage], stop: Optional[List[str]] = None
    ) -> ChatResult:
        pass
    
    def _create_message_dicts(
        self, 
        messages: List[BaseMessage], 
        stop: Optional[List[str]]
    ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
        params = self._default_params
        if stop is not None:
            if "stop" in params:
                raise ValueError("`stop` found in both the input and default params.")
            params["stop"] = stop
        message_dicts = [self._convert_message_to_dict(m) for m in messages]
        return message_dicts, params

• 將我們轉換成 LangChain 框架的台智雲 Model 存成一個 py 檔，就可以像前面 OpenAI 和 Gemini 那樣呼叫語言模型

from langchain_ffm import ChatFormosaFoundationModel

llm = ChatFormosaFoundationModel(model="ffm-llama3-70b-chat", temperature=0.01)
response = llm.invoke("Aespa的成員有誰？")
print(response.content)

程式碼結果探討 🧐：

• 看到這邊不覺得有異曲同工之妙嗎，這就是 LangChain 架構迷人的地方！
• 封裝成 LangChain 那邊，我大概看得懂在幹嘛，但要我詳細解釋的部分是有點難，因為我也是參考 台智雲 Conversation API 的 code 做微調改成我要的樣子。
• 我的理解是，簡單來說第一個 class 就是定義模型一些基本功能和呼叫功能。第二個 class 就是基於框架來幫助我來跟模型做互動，算是與模型互動的橋樑

結論

今天簡單了解了 LangChain 框架是什麼，然後如何透過 LangChain 框架使用 LLMs。那後面會繼續來分享更多的 LangChain 框架的應用，希望可以讓原本覺得 LangChain 很麻煩的人或者覺得 LangChain 框架很複雜的人知道 LangChain 的好，因為目前主流的服務 LangChain 都有整合進框架之中。

題外話🤣

悠閒的 Happy Friday，雖然沒有放假，但可以做自己的事。我就 focus 在鐵人賽文章，這樣晚上可以跟女朋友去約會了🥰。啊我同事玩了一整個下午的 LOL 和全明星街球派對哈哈哈，難得有這麼清閒的下午，還有薪水可以拿，但是他一直輸是蠻好笑的哈哈哈

下一篇文章：LangChain 怎麼 Chain？